package com.geomesa.spark.SparkCore

package com.geomesa.spark.SparkCore

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{SparkSession, types}
import org.geotools.geojson.feature.FeatureJSON
import org.geotools.geojson.geom.GeometryJSON
import org.locationtech.jts.geom.MultiLineString

import scala.reflect.internal.util.Origins.MultiLine

/**
  * 构建UDF的方式
  */
object SparkSqlFun {

  def main(args: Array[String]): Unit = {

    import org.locationtech.geomesa.spark.jts._
    //spark
    val spark: SparkSession = {
      SparkSession.builder()
        .appName("ByUdfFunction")
        .master("local[*]")
        .getOrCreate()
        //需注入spark.jts._包
        .withJTS
    }

    //UDF
    val schema = StructType(Array(
      StructField("id", StringType),
      //显示路线名称
      /*       StructField("properties",StructType(Array(
               StructField("NAME1",StringType)
             ))),*/
      StructField("geometry", StructType(Array(
        StructField("type", StringType),
        StructField("coordinates", types.ArrayType(StringType))))
        //StructField("coordinates", types.ArrayType(StringType))))
        //StructField("coordinates",  types.ArrayType(DoubleType))))
      )))

    val schema1 = StructType(Array(
      StructField("id",StringType),
      StructField("geometry",StringType)
    ))

    val dataFile = this.getClass.getClassLoader.getResource("line.txt").getPath
    val df = spark.read
      .schema(schema)
      .json(dataFile)
    //.show(5)
    //.printSchema()


    import spark.implicits._
    val alertDf = df.withColumn("geometryType", $"geometry.type")
      .withColumn("coordinates", $"geometry.coordinates")
    //.show(5)

    val geojson = new GeometryJSON()

    val sc = alertDf.select($"geometry").show(5)
    val mls = geojson.readMultiLine(sc)
    println(mls.asInstanceOf[MultiLineString])
    import spark.sql


    /* alertDf.createOrReplaceTempView("coordinate")
     import spark.sql
     val de = sql("select geometry,type,coordinates,'LINESTRING (0 2,0 0)' as line from coordinate")
     de.withColumn("pointA", st_makePoint(121.97738334,37.47710858))
       .withColumn("pointB", st_makePoint(121.97738305,37.47710848))
       //.withColumn("line",st_makeLine())
       .withColumn("lineString", st_lineFromText($"line"))
       .withColumn("distance", st_distance($"pointA", $"lineString"))
       .withColumn("pointDistance", st_distance($"pointA", $"pointB"))
       .show(5)*/

    /*    root
        |-- geometry: struct (nullable = true)
        |    |-- coordinates: array (nullable = true)
        |    |    |-- element: array (containsNull = true)
        |    |    |    |-- element: array (containsNull = true)
        |    |    |    |    |-- element: double (containsNull = true)
        |    |-- type: string (nullable = true)
        |-- geometry_name: string (nullable = true)*/


    /*+--------+--------------------+---------------+--------------------+
    |      id|            geometry|           type|         coordinates|
      +--------+--------------------+---------------+--------------------+
    |sdgsmc.1|[MultiLineString,...|MultiLineString|[[[1.357846020181...|
    |sdgsmc.2|[MultiLineString,...|MultiLineString|[[[1.319414533241...|
    |sdgsmc.3|[MultiLineString,...|MultiLineString|[[[1.287772131522...|
    |sdgsmc.4|[MultiLineString,...|MultiLineString|[[[1.319625761819...|
    |sdgsmc.5|[MultiLineString,...|MultiLineString|[[[1.322615028157...|
    +--------+--------------------+---------------+--------------------+*/

  }
}

